package com.yanggu.spark.core.demand.categoryTop10

import org.apache.spark.{SparkConf, SparkContext}

//这里使用累加器进行优化
object CategoryCountTop10_2 {

  def main(args: Array[String]): Unit = {

    //1. 创建spark配置对象
    val sparkConf = new SparkConf().setAppName("CategoryCountTop102").setMaster("local[*]")

    //2. 创建spark上下文环境对象
    val sparkContext = new SparkContext(sparkConf)

    //3. 初始化累加器
    val categoryCountAccumulator = new CategoryCountAccumulator

    //4. 注册累加器
    sparkContext.register(categoryCountAccumulator, "categoryCountAccumulator")

    //5. 读取文件
    sparkContext.textFile("input/user_visit_action.txt")
      //6. 循环遍历使用累加器进行聚合运算
      .foreach(line => {
        //使用模式匹配
        line.split("_") match {
          //计算点击次数
          case Array(_, _, _, _, _, "null", click_category_id, _, "null", "null", "null", "null", _) =>
            categoryCountAccumulator.add((click_category_id, "click"))
          //计算下单次数
          case Array(_, _, _, _, _, "null", "-1", "-1", order_category_ids, _, "null", "null", _) =>
            order_category_ids.split(",").map(id => categoryCountAccumulator.add((id, "order")))
          //计算支付次数
          case Array(_, _, _, _, _, "null", "-1", "-1", "null", "null", pay_category_ids, _, _) =>
            pay_category_ids.split(",").map(id => categoryCountAccumulator.add((id, "pay")))
          //如果是其他, 什么都不做
          case _ =>
        }
      })

    //累加器最后是将各个task中累加器的值调用merge方法进行合并, 最后将数据拉取到driver端。
    //因此如果累加器最后合并的数量过大, 可能会导致堆内存溢出。这点在使用时需要仔细考虑数据的规模
    //6. 获取累加器的值。并且排序, 得到top10
    categoryCountAccumulator
      .value
      .toList
      //这里使用tuple进行排序, 默认按照升序排序, 可以实现类似于sql中类似的功能。
      //例如先按照A升序, 然后按照B降序。如果要降序, 可以对值进行加上一个 "-" 号进行取反, 表示降序
      .sortBy {
        case (_, categoryCount) => (-categoryCount.clickCount, -categoryCount.orderCount, -categoryCount.payCount)
      }
      .take(10)
      .foreach(println)

  }

}
